package com.atguigu.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp extends BaseApp {
    public static void main(String[] args) {
        new BaseLogApp().start(
                10002,
                4,
                "base_log_app",
                Constant.TOPIC_LOG
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //kafkaStrDS.print("kafkaSource");
        //清洗数据,将不完整的json数据输出到侧输出流,并将流中的jsonStr转化成jsonObj
        OutputTag<String> dirtyDSTag
                = new OutputTag<String>("dirtyData", Types.STRING){};
        SingleOutputStreamOperator<JSONObject> cleanedDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(jsonStr);
                            out.collect(jsonObject);
                        } catch (Exception e) {
                            ctx.output(dirtyDSTag, jsonStr);
                        }
                    }
                }
        );
        SideOutputDataStream<String> dirtyDS = cleanedDS.getSideOutput(dirtyDSTag);
        //cleanedDS.print("cleanedDS");
        //dirtyDS.print("dirtyDS");
        //将脏数据写入到dirty_data侧输出流
        KafkaSink<String> dirtyDataSink = FlinkSinkUtil.getKafkaSink("dirty_data");
        dirtyDS.sinkTo(dirtyDataSink);

        //新老访客标记修复
        //按照mid进行分组
        KeyedStream<JSONObject, String> keyByMidDS = cleanedDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        //新老访客标记修复
        SingleOutputStreamOperator<JSONObject> fixedDS = keyByMidDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> valueState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<>("firstLoginState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //获取状态中的上次访问时间
                        String firstLoginDate = valueState.value();
                        Long ts = jsonObject.getLong("ts");
                        String dt = DateFormatUtil.tsToDate(ts);
                        String isNew = jsonObject.getJSONObject("common").getString("is_new");
                        //所有界面都没有访问过
                        if ("1".equals(isNew)){
                            //状态中的最后访问日期为空，则为首次访问的访客
                            if (firstLoginDate==null){
                                valueState.update(dt);
                            }else {
                                //如果最后访问日期不为空，则认为是老访客，清楚过数据
                                if (!firstLoginDate.equals(dt)){
                                    isNew = "0";
                                    jsonObject.getJSONObject("common").put("is_new",isNew);
                                }
                            }
                        }else {  //所有界面都访问过
                            if (firstLoginDate==null){
                                String yesterday = DateFormatUtil.tsToDate(ts-1000*60*60*24);
                                valueState.update(yesterday);
                            }
                        }
                        out.collect(jsonObject);
                    }
                }
        );
        //fixedDS.print("fixedDS");
        //分流
        //定义启动、曝光、动作、错误、播放侧输出流标签
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideoTag") {};

        //开始分流
        SingleOutputStreamOperator<String> separatedDS = fixedDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, String>.Context context, Collector<String> out) throws Exception {
                        // 5.2.1 收集错误数据
                        JSONObject error = jsonObj.getJSONObject("err");
                        if (error != null) {
                            context.output(errorTag, jsonObj.toJSONString());
                        }

                        // 剔除 "err" 字段
                        jsonObj.remove("err");

                        // 5.2.2 收集启动数据
                        JSONObject start = jsonObj.getJSONObject("start");
                        if (start != null) {
                            context.output(startTag, jsonObj.toJSONString());
                        } else {
                            // 获取 "common" 字段
                            JSONObject common = jsonObj.getJSONObject("common");
                            // 获取 "ts"
                            Long ts = jsonObj.getLong("ts");
                            JSONObject appVideo = jsonObj.getJSONObject("appVideo");

                            //收集播放数据
                            if (appVideo != null) {
                                context.output(appVideoTag, jsonObj.toJSONString());
                            } else {

                                // 获取 "page" 字段
                                JSONObject page = jsonObj.getJSONObject("page");

                                //收集曝光数据
                                JSONArray displays = jsonObj.getJSONArray("displays");
                                if (displays != null) {
                                    for (int i = 0; i < displays.size(); i++) {
                                        JSONObject display = displays.getJSONObject(i);
                                        JSONObject displayObj = new JSONObject();
                                        displayObj.put("display", display);
                                        displayObj.put("common", common);
                                        displayObj.put("page", page);
                                        displayObj.put("ts", ts);
                                        context.output(displayTag, displayObj.toJSONString());
                                    }
                                }

                                //收集动作数据
                                JSONArray actions = jsonObj.getJSONArray("actions");
                                if (actions != null) {
                                    for (int i = 0; i < actions.size(); i++) {
                                        JSONObject action = actions.getJSONObject(i);
                                        JSONObject actionObj = new JSONObject();
                                        actionObj.put("action", action);
                                        actionObj.put("common", common);
                                        actionObj.put("page", page);
                                        context.output(actionTag, actionObj.toJSONString());
                                    }
                                }

                                //收集页面数据
                                jsonObj.remove("displays");
                                jsonObj.remove("actions");
                                out.collect(jsonObj.toJSONString());
                            }
                        }

                    }
                }
        );
        // 打印主流和各侧输出流查看分流效果
        separatedDS.print("page");
        separatedDS.getSideOutput(startTag).print("start");
        separatedDS.getSideOutput(displayTag).print("display");
        separatedDS.getSideOutput(actionTag).print("action");
        separatedDS.getSideOutput(errorTag).print("error");
        separatedDS.getSideOutput(appVideoTag).print("appVideo");

        DataStream<String> startDS = separatedDS.getSideOutput(startTag);
        DataStream<String> displayDS = separatedDS.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedDS.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedDS.getSideOutput(errorTag);
        DataStream<String> appVideoDS = separatedDS.getSideOutput(appVideoTag);

        separatedDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        startDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        displayDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        actionDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        errorDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        appVideoDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_APPVIDEO));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
